package com.apps.sdses.flink141.jar;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;
import java.util.LinkedList;

/**
 * author: codejiwei
 * date: 2023/8/14
 * desc: flink tumble window update timestamp field
 **/
public class FlinkTumbleWindowUpdateTs {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Schema schema = Schema.newBuilder()
                .column("id", DataTypes.INT())
                .column("name", DataTypes.STRING())
                .column("price", DataTypes.INT())
                .column("ts", DataTypes.TIMESTAMP(3))
                .watermark("ts", "ts - INTERVAL '3' SECOND")
                .build();
        Table table = tableEnv.from(
                TableDescriptor.forConnector("datagen")
                        .schema(schema)
                        .option("rows-per-second", "1")
//                        .option("number-of-rows", "10")
                        .option("fields.name.length", "5")
                        .option("fields.id.min", "0")
                        .option("fields.id.max", "10")
                        .option("fields.price.min", "1")
                        .option("fields.price.max", "10")
                        .build()
        );

        tableEnv.toDataStream(table)
                //转为tuple key by window
//                .map(x -> Tuple4.of(x.<Integer>getFieldAs("id"), x.<String>getFieldAs("name"), x.<Integer>getFieldAs("price"), x.<LocalDateTime>getFieldAs("ts")))
//                .returns(Types.TUPLE(Types.INT, Types.STRING, Types.INT, Types.LOCAL_DATE_TIME))
//                .keyBy(x -> x.f0)
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .process(new ProcessWindowFunction<Tuple4<Integer, String, Integer, LocalDateTime>, String, Integer, TimeWindow>() {
//                    @Override
//                    public void process(Integer integer, Context context, Iterable<Tuple4<Integer, String, Integer, LocalDateTime>> elements, Collector<String> out) throws Exception {
//                        long start = context.window().getStart();
//                        long end = context.window().getEnd();
//                        for (Tuple4<Integer, String, Integer, LocalDateTime> element : elements) {
//                            out.collect("window start:" + start + ", window end:" + end + ", value:" + element);
//                        }
//                    }
//                })
//                .sum(2)
//                .print();

                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessAllWindowFunction<Row, Row, TimeWindow>() {
                    private ListState<LocalDateTime> listState;
                    private ReducingState<LocalDateTime> reducingState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ListStateDescriptor<LocalDateTime> listStateDescriptor = new ListStateDescriptor<>("listState", LocalDateTime.class);
                        listState = getRuntimeContext().getListState(listStateDescriptor);

                        ReducingStateDescriptor<LocalDateTime> descriptor = new ReducingStateDescriptor<>("reduceState", new ReduceFunction<LocalDateTime>() {
                            @Override
                            public LocalDateTime reduce(LocalDateTime value1, LocalDateTime value2) throws Exception {
                                return value1.isBefore(value2) ? value1 : value2;
                            }

                        }, LocalDateTime.class);
                        reducingState = getRuntimeContext().getReducingState(descriptor);
                    }

                    @Override
                    public void clear(Context context) throws Exception {
                        reducingState.clear();
                    }

                    @Override
                    public void process(Context context, Iterable<Row> elements, Collector<Row> out) throws Exception {

                        LinkedList<LocalDateTime> localDateTimes = new LinkedList<>();

                        LocalDateTime lastVisitTime = null;

                        //添加时间字段到state
//                        for (Row element : elements) {
//                            System.out.println("========" + element);
//                            LocalDateTime ts = element.<LocalDateTime>getFieldAs("ts");
//                            reducingState.add(ts);
//                        }
//
//                        for (Row element : elements) {
//                            element.setField("ts", reducingState.get());
//                            out.collect(element);
//                        }

                        //
                        for (Row element : elements) {
                            System.out.println("========" + element);
                            LocalDateTime ts = element.<LocalDateTime>getFieldAs("ts");
                            lastVisitTime = ts;
                        }
                        for (Row element : elements) {
                            element.setField("ts", lastVisitTime);
                            out.collect(element);
                        }

//                        for (Row element : elements) {
//                            LocalDateTime ts = element.<LocalDateTime>getFieldAs("ts");
//                            listState.add(ts);
//                            out.collect(element);
//                        }
//                        for (Row element : elements) {
//                            LocalDateTime ts = element.<LocalDateTime>getFieldAs("ts");
//                            listState.add(ts);
//                            localDateTimes.add(ts);
//                        }
//                        for (Row element : elements) {
//
//                            element.setField("ts", localDateTimes.getLast());
//                            out.collect(element);
//                        }
                        Thread.sleep(1000);
                    }
                })
                .print();
        env.execute();

    }

}
